package com.sbtr.business.publishtask.process;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.sbtr.business.publishtask.entity.PublishPredictTask;
import com.sbtr.business.publishtask.entity.PublishTaskData;
import com.sbtr.business.publishtask.service.impl.PublishPredictTaskServiceImpl;
import com.sbtr.business.trainingmodel.entity.TrainingModelData;
import com.sbtr.business.trainingmodel.service.impl.TrainingModelDataServiceImpl;
import com.sbtr.business.trainingserver.entity.TrainingServer;
import com.sbtr.business.trainingserver.service.IPublishServerHttpApi;
import com.sbtr.open.publishtask.dto.PublishPredictTaskDto;
import com.sbtr.open.trainingserver.dto.TrainingServerDto;
import com.sbtr.util.BeanCopyUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

//import com.sbtr.util.MinioUtil;
//import com.sbtr.util.OSSUtil;

@Service
public class PredictDataMultiProcessor {
    private static final Logger logger = LoggerFactory.getLogger(PredictDataMultiProcessor.class);

    @Autowired
    private PublishPredictTaskServiceImpl publishPredictTaskService;

    @Autowired
    private TrainingModelDataServiceImpl trainingModelService;

    @Autowired
    private IPublishServerHttpApi publishServerHttpApi;

//	@Autowired
//	private OSSUtil ossUtil;

//	@Autowired
//	private MinioUtil minioUtil;

    private String bucketName = "meicloud-ai-platform";

    @Value("${predict-data-task.branch}")
    private int predictBranchValue;

    @Value("${predict-data-task.thread-num}")
    private int threadNum;

    @Value("${predict-data-task.retry-max}")
    private int retryMax;

    @Value("${oss.config.downloadPath}")
    private String downloadPath;


	/*

	public List<String[]> readFileData(PublishPredictTask task) {
		String fileName = task.getPredictFile();
		if (StringUtils.isBlank(fileName)) {
			task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
			task.setError("文件名不能为空");
			task.setEndTime(new Date());
			publishPredictTaskService.saveOrUpdate(task);
			throw new RuntimeException("文件名不能为空");
		}

		File downloadFile = null;
		try {
			downloadFile = ossUtil.downloadFile(fileName);
		} catch (Exception e) {
			logger.error(fileName + " oss下载文件失败", e);
			e.printStackTrace();
			task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
			task.setError(fileName + " oss下载文件失败");
			task.setEndTime(new Date());
			publishPredictTaskService.saveOrUpdate(task);
			throw new RuntimeException(fileName + " oss下载文件失败");
		}

		if (downloadFile == null) {
			logger.error(fileName + " 获取失败");
			task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
			task.setError(fileName + " 获取失败");
			task.setEndTime(new Date());
			publishPredictTaskService.saveOrUpdate(task);
			throw new RuntimeException(fileName + " 获取失败, 请联系管理员或重试");
		}

		List<String[]> predictData = null;
		if (fileName.toLowerCase().endsWith(".xls") || fileName.toLowerCase().endsWith(".xlsx")) {
			try {
				predictData = ExcelReader.read(downloadFile);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				logger.error(fileName + " 读取文件失败", e);
				e.printStackTrace();
				task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
				task.setError(fileName + " 读取文件失败");
				task.setEndTime(new Date());
				publishPredictTaskService.saveOrUpdate(task);
				throw new RuntimeException(fileName + " 读取文件失败");
			}
		} else {
			try {
				predictData = new ArrayList<String[]>();
				InputStreamReader reader = new InputStreamReader(new FileInputStream(downloadFile));
				BufferedReader br = new BufferedReader(reader);
				String line = null;
				while ((line = br.readLine()) != null) {
					if (fileName.toLowerCase().endsWith(".csv")) {
						predictData.add(line.split(","));
					} else {

						predictData.add(line.split("[\t]"));
					}
				}
				br.close();
				reader.close();
			} catch (Exception e) {
				logger.error(fileName + " 读取文件失败", e);
				e.printStackTrace();
				task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
				task.setError(fileName + " 读取文件失败");
				task.setEndTime(new Date());
				publishPredictTaskService.saveOrUpdate(task);
				throw new RuntimeException(fileName + " 读取文件失败");
			}
		}
		return predictData;
	}*/

	/*
	public List<String[]> readFileDataFromMinio(PublishPredictTask task, PredictDataMultiNewDto dto) {
		String fileName = task.getPredictFile();
		if (StringUtils.isBlank(fileName)) {
//			task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
			task.setError("文件名不能为空");
			task.setEndTime(new Date());
			publishPredictTaskService.saveOrUpdate(task);
			throw new RuntimeException("文件名不能为空");
		}

		File downloadFile = null;

		try {
//			downloadFile = ossUtil.downloadFile(fileName);
			String bucketname=dto.getBucketDir();
			String fileNames=bucketname.substring(bucketname.indexOf("/")+1,bucketname.length())+"/"+dto.getFileOssName();
			downloadFile =  minioUtil.getObject(bucketName,fileNames,downloadPath,fileName);
		} catch (Exception e) {
			logger.error(fileName + " oss下载文件失败", e);
			e.printStackTrace();
			task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
			task.setError(fileName + " oss下载文件失败");
			task.setEndTime(new Date());
			publishPredictTaskService.saveOrUpdate(task);
			throw new RuntimeException(fileName + " oss下载文件失败");
		}

		if (downloadFile == null) {
			logger.error(fileName + " 获取失败");
			task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
			task.setError(fileName + " 获取失败");
			task.setEndTime(new Date());
			publishPredictTaskService.saveOrUpdate(task);
			throw new RuntimeException(fileName + " 获取失败, 请联系管理员或重试");
		}

		List<String[]> predictData = null;
		if (fileName.toLowerCase().endsWith(".xls") || fileName.toLowerCase().endsWith(".xlsx")) {
			try {
				predictData = ExcelReader.read(downloadFile);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				logger.error(fileName + " 读取文件失败", e);
				e.printStackTrace();
				task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
				task.setError(fileName + " 读取文件失败");
				task.setEndTime(new Date());
				publishPredictTaskService.saveOrUpdate(task);
				throw new RuntimeException(fileName + " 读取文件失败");
			}
		} else {
			try {
				predictData = new ArrayList<String[]>();
				InputStreamReader reader = new InputStreamReader(new FileInputStream(downloadFile));
				BufferedReader br = new BufferedReader(reader);
				String line = null;
				while ((line = br.readLine()) != null) {
					if (fileName.toLowerCase().endsWith(".csv")) {
						predictData.add(line.split(","));
					} else {

						predictData.add(line.split("[\t]"));
					}
				}
				br.close();
				reader.close();
			} catch (Exception e) {
				logger.error(fileName + " 读取文件失败", e);
				e.printStackTrace();
				task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
				task.setError(fileName + " 读取文件失败");
				task.setEndTime(new Date());
				publishPredictTaskService.saveOrUpdate(task);
				throw new RuntimeException(fileName + " 读取文件失败");
			}
		}
		return predictData;
	}
*/


    public List<String[]> predict(PublishPredictTask task, PublishTaskData publishTask, TrainingServer server,
                                  List<String[]> predictData) {
        Long publishId = task.getPublishId();

        String[] title = predictData.get(0);
        predictData.remove(0);

        // 获取数据对应的index
        Long modelId = publishTask.getModelId();
        TrainingModelData trainingModel = trainingModelService.getById(modelId);
        String maps = trainingModel.getClassifier();
        JSONObject paramsMap = JSONObject.parseObject(maps);
        JSONObject featureJSONObject = paramsMap.getJSONObject("feature");
        JSONArray featureNameJSONArray = paramsMap.getJSONArray("feature_name");

        List<String> featureNameList = featureNameJSONArray.toJavaList(String.class);
        Map<String, Integer> variableIndexMap = new HashMap<>();
        for (String variable : featureNameList) {
            for (int index = 0; index < title.length; index++) {
                if (variable.equals(title[index])) {
                    variableIndexMap.put(variable, index);
                }
            }
        }

        int variableValuedSize = variableIndexMap.size();
        if (variableValuedSize != featureNameList.size()) {
            task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
            task.setError("文件格内容格式出错，请检查，txt请使用 [\\t]作为分割符，csv请使用[,]作为分隔符 !");
            task.setEndTime(new Date());
            publishPredictTaskService.saveOrUpdate(task);
            throw new RuntimeException("文件格内容格式出错，请检查，txt请使用 [\\t]作为分割符，csv请使用[，]作为分隔符 !");
        }
        ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
        List<Future<List<String[]>>> predictDataResultFuture = new ArrayList<>();
        int predictDataRemainder = predictData.size() % predictBranchValue;
        int predictDataBranch = predictData.size() / predictBranchValue;
        double precentBranch = predictBranchValue * threadNum * 2;

        if (predictDataRemainder != 0) {
            predictDataBranch += 1;
        }

        boolean hasError[] = {false};
        for (int i = 0; i < predictDataBranch; i++) {
            // for (String[] data : predictData) {
            int num = i;
            double pBranch = precentBranch;
            if (hasError[0]) {
                executorService.shutdown();
                logger.error("预测过程中出现错误了, 预测已经提前结束");
                break;
            }

            Future<List<String[]>> future = executorService.submit(new Callable<List<String[]>>() {
                public List<String[]> call() throws Exception {
                    int startNum = num * predictBranchValue;
                    int endNum = (num + 1) * predictBranchValue;

                    if (endNum > predictData.size()) {
                        endNum = predictData.size();
                    }

                    if (startNum % pBranch == 0) {
                        double preidctPrecent = startNum / (double) task.getTotalRowData();
                        task.setPredictPrecent(preidctPrecent);
                        publishPredictTaskService.saveOrUpdate(task);
                    }

                    JSONArray pData = new JSONArray();
                    List<String[]> pBranch = new ArrayList<>();
                    for (int k = startNum; k < endNum; k++) {
                        String[] data = predictData.get(k);
                        pBranch.add(data);
                        if (data.length != title.length) {
                            hasError[0] = true;
                            task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
                            task.setError("文件内容格式出错，请检查!");
                            task.setEndTime(new Date());
                            publishPredictTaskService.saveOrUpdate(task);
                            logger.error("文件内容格式出错，请检查!");
                            throw new RuntimeException("文件内容格式出错，请检查!");
                        }
                        JSONObject params = new JSONObject();
                        for (String variable : featureNameList) {
                            int index = variableIndexMap.get(variable);
                            if (index > data.length) {
                                hasError[0] = true;
                                task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
                                task.setError("预测数据存在问题，请检查");
                                task.setEndTime(new Date());
                                logger.error("预测数据存在问题，请检查");
                                publishPredictTaskService.saveOrUpdate(task);
                                throw new RuntimeException("预测数据存在问题，请检查");
                            } else {
                                if (featureJSONObject.containsKey(variable)) {
                                    params.put(variable, data[index]);
                                } else {
                                    try {
                                        params.put(variable, new Double(data[index]));
                                    } catch (Exception e) {
                                        hasError[0] = true;
                                        e.printStackTrace();
                                        task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
                                        task.setError("预测数据存在问题，请检查");
                                        task.setEndTime(new Date());
                                        logger.error("预测数据存在问题，请检查");
                                        publishPredictTaskService.saveOrUpdate(task);
                                        throw new RuntimeException("预测数据存在问题，请检查");
                                    }
                                }

                            }
                        }

                        pData.add(params);

                    }
                    JSONArray predictResult = null;

                    try {
                        predictResult = predictWithRetry(BeanCopyUtils.copyProperties(server, TrainingServerDto.class),
                                publishId, pData);
                    } catch (Exception e) {
                        hasError[0] = true;
                        e.printStackTrace();
                        task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
                        task.setError(e.getMessage());
                        task.setEndTime(new Date());
                        publishPredictTaskService.saveOrUpdate(task);
                        throw e;
                    }

                    List<String[]> resultList = new ArrayList<>();
                    for (int ri = 0; ri < predictResult.size(); ri++) {
                        String[] data = pBranch.get(ri);
                        String pResult = predictResult.get(ri).toString();
                        List<String> tmpList = new ArrayList<String>();
                        tmpList.addAll(Arrays.asList(data));
                        tmpList.add(pResult);
                        String[] tmpString = tmpList.toArray(new String[tmpList.size()]);
                        resultList.add(tmpString);
                    }

                    return resultList;
                }
            });
            predictDataResultFuture.add(future);
        }

        executorService.shutdown();

        List<String[]> predcitDataResult = new ArrayList<>();

        List<Boolean> predictDataResultBoolean = predictDataResultFuture.stream().map(x -> {
            try {
                predcitDataResult.addAll(x.get());
                return true;
            } catch (InterruptedException | ExecutionException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
                return false;
            }
        }).collect(Collectors.toList());

        if (predcitDataResult == null || predcitDataResult.size() != predictData.size()) {
            task.setStatus(PublishPredictTaskDto.PREDICT_ERROR);
            task.setError("预测结果提取出现了问题，请检查预测数据或重试");
            task.setEndTime(new Date());
            logger.error("预测结果提取出现了问题，请检查预测数据或重试");
            publishPredictTaskService.saveOrUpdate(task);
            throw new RuntimeException("预测结果提取出现了问题，请检查预测数据或重试");
        }

        List<String> tmpTitle = new ArrayList<String>();
        tmpTitle.addAll(Arrays.asList(title));
        tmpTitle.add("predict");
        predcitDataResult.add(0, tmpTitle.toArray(new String[tmpTitle.size()]));
        return predcitDataResult;
    }

    /**
     * 带重试的预测
     */
    public JSONArray predictWithRetry(TrainingServerDto dto, Long publishId, JSONArray pData) throws Exception {
        int tryTimes = 0;
        JSONArray predictResult = null;
        Exception except = null;
        while (tryTimes < retryMax) {
            try {
                //	predictResult = publishServerHttpApi.predictData(dto, publishId, pData);
                break;
            } catch (Exception e) {
                except = e;
                tryTimes++;
            }
        }

        if (tryTimes < retryMax) {
            return predictResult;
        } else {
            throw except;
        }
    }
}
